package com.tzy.rpc.loadbalancer;/**
 * @description:
 * @Author HP
 * @create 2022/4/23 10:14
 */

/**
 * @author HP
 * @create 2022-04-23 10:14
 * @desc
 **/

import com.alibaba.nacos.api.naming.pojo.Instance;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * refer to dubbo consistent hash load balance: https://github.com/apache/dubbo/blob/2d9583adf26a2d8bd6fb646243a9fe80a77e65d5/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalance.java
 *
 * @author RicardoZ
 * @createTime 2020年10月20日 18:15:20
 */

@Slf4j
public class ConsistentHashLoadBalance implements LoadBalancer {

    /*
    Nacos的instance数据格式：
    Instance{instanceId='127.0.0.1#9997#DEFAULT#DEFAULT_GROUP@@com.tzy.rpc.api.HelloService', ip='127.0.0.1', port=9997, weight=1.0, healthy=true, enabled=true, ephemeral=true, clusterName='DEFAULT', serviceName='DEFAULT_GROUP@@com.tzy.rpc.api.HelloService', metadata={}}
    1.拉取全部的服务存起来
    2.要有虚拟节点和真实节点
    */

    /*
    物理节点
     */
    private static Set<String> PHYSICAL_NODES = new TreeSet<>();
    private ConcurrentHashMap<String,Instance> dic=new ConcurrentHashMap<>();

    //虚拟节点
    private final int VIRTUAL_COPIES = 1048576; // 物理节点至虚拟节点的复制倍数
    private TreeMap<Long, String> virtualNodes = new TreeMap<>(); // 哈希值 => 物理节点

    // 32位的 Fowler-Noll-Vo 哈希算法
    // https://en.wikipedia.org/wiki/Fowler–Noll–Vo_hash_function
    private static Long FNVHash(String key) {
        final int p = 16777619;
        Long hash = 2166136261L;
        for (int idx = 0, num = key.length(); idx < num; ++idx) {
            hash = (hash ^ key.charAt(idx)) * p;
        }
        hash += hash << 13;
        hash ^= hash >> 7;
        hash += hash << 3;
        hash ^= hash >> 17;
        hash += hash << 5;

        if (hash < 0) {
            hash = Math.abs(hash);
        }
        return hash;
    }

    public ConsistentHashLoadBalance() {
    }

    // 根据物理节点，构建虚拟节点映射表
    public ConsistentHashLoadBalance(List<Instance> instances) {
        for (Instance instance : instances) {
            PHYSICAL_NODES.add(instance.getPort()+"");
            dic.put(instance.getPort()+"",instance);
        }

        for (String nodeIp : PHYSICAL_NODES) {
            addPhysicalNode(nodeIp);
        }
    }

    // 添加虚拟节点
    public void addPhysicalNode(String nodeIp) {
        String port=nodeIp;
        for (int idx = 0; idx < VIRTUAL_COPIES; ++idx) {
            long hash = FNVHash(port+"#"+idx);//port#idx(1048576)
            virtualNodes.put(hash, nodeIp);
        }
    }

    // 删除虚拟节点
    public void removePhysicalNode(String nodeIp) {
        System.out.println("移除port： "+nodeIp);
        for (int idx = 0; idx < VIRTUAL_COPIES; ++idx) {
            long hash = FNVHash(nodeIp + "#" + idx);
            virtualNodes.remove(hash);
        }
    }

    // 查找对象映射的节点
    public String getObjectNode(String object) {
        long hash = FNVHash(object);
        SortedMap<Long, String> tailMap = virtualNodes.tailMap(hash); // 所有大于 hash 的节点
        Long key = tailMap.isEmpty() ? virtualNodes.firstKey() : tailMap.firstKey();
        return virtualNodes.get(key);
    }

    // 统计对象与节点的映射关系(测试用)
    public void dumpObjectNodeMap(String label, int objectMin, int objectMax) {
        // 统计
        Map<String, Integer> objectNodeMap = new TreeMap<>(); // port => COUNT
        for (int object = objectMin; object <= objectMax; ++object) {
            String nodeIp = getObjectNode(Integer.toString(object));
            Integer count = objectNodeMap.get(nodeIp);
            objectNodeMap.put(nodeIp, (count == null ? 0 : count + 1));
        }

        // 打印
        double totalCount = objectMax - objectMin + 1;
        System.out.println("======== " + label + " ========");
        for (Map.Entry<String, Integer> entry : objectNodeMap.entrySet()) {
            long percent = (int) (100 * entry.getValue() / totalCount);
            System.out.println("IP=" + entry.getKey() + ": RATE=" + percent + "%");
        }
    }


    @Override
    public Instance select(List<Instance> instances) {
        int idx = new Random().nextInt(16777619);
        String port = getObjectNode(idx+"");
        System.out.println("本次选择的是port："+port);
        if(dic.get(port).isHealthy()){
            return dic.get(port);
        }else {
            System.out.println("port不可用");
            removePhysicalNode(port);
            dic.remove(port);
            PHYSICAL_NODES.remove(port);
            return select(instances);
        }
    }
}

